package mapreduce;

import java.nio.ByteBuffer;

import mapreduce.Record.RecordData;

public abstract class BytesFormat {

	static BytesFormat getFormt(int code) {
		int formatNum = m_formats.length;
		for (int i = 0; i < formatNum; i++) {
			if (m_formats[i].getCode() == code) {
				return (BytesFormat) m_formats[i].getFormat();
			}
		}
		return null;
	}
	
	public abstract BytesFormat getFormat();

	/**
	 * from non-readable format to object array
	 * 
	 * @param buffer
	 * @return
	 */
	static public RecordData decodeBuffer(byte[] bytes) {
		ByteBuffer buffer = ByteBuffer.wrap(bytes);
		int code = buffer.getInt();
		BytesFormat format = getFormt(code);
		Object obj = format.decode(buffer);
		return new RecordData(obj, format);
	}

	public static byte[] encodeBuffer(RecordData recordData) {
		ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
		BytesFormat format = recordData.getFormat();
		buffer.putInt(format.getCode());
		buffer = format.encode(recordData.getData(), buffer);
		byte[] bytes = new byte[buffer.position()];
		buffer.rewind();
		buffer.get(bytes);
		buffer = null;
		return bytes;
	}
	
	static private int BUFFER_SIZE = 128;

	public static byte[] formatBuffer(RecordData recordData) {
		return recordData.getFormat().format(recordData.getData());
	}

	abstract protected Object decode(ByteBuffer buffer);

	abstract protected ByteBuffer encode(Object object, ByteBuffer buffer);

	abstract protected int getCode() ;

	public boolean equals(Object anotherObject){
		BytesFormat anotherFormat = (BytesFormat) anotherObject;
		if(getCode() == anotherFormat.getCode()){
			return true;
		}
		else{
			return false;
		}
	}
	/**
	 * from non-readable format to readable format
	 * 
	 * @param buffer
	 * @return
	 */
	protected byte[] format(Object object) {
		return object.toString().getBytes();
	}

	static BytesFormat[] m_formats = { new IntegerFormat(), new StringFormat(),
			new MultipleTypeFormat(), new DynamicNumberFormat(), new NullFormat() };
	
	static public void main(String[] args){
		System.out.println(Integer.class.hashCode());
	}
}
